package com.yu.dao;

import com.yu.bean.Message;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.HBaseConfTool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author wu
 * @create 2022-10-14-17:58
 */
public class HbaseDao {


    private Configuration conf = HBaseConfiguration.create();
    //微博内容表的表名
    private static final byte[] TABLE_CONTENT =
            Bytes.toBytes("weibo:content");
    //用户关系表的表名
    private static final byte[] TABLE_RELATIONS =
            Bytes.toBytes("weibo:relations");
    //微博收件箱表的表名
    private static final byte[] TABLE_RECEIVE_CONTENT_EMAIL =
            Bytes.toBytes("weibo:receive_content_email");

    public void initNamespace() {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
//命名空间类似于关系型数据库中的 schema，可以想象成文件夹
            NamespaceDescriptor weibo = NamespaceDescriptor
                    .create("weibo")
                    .addConfiguration("creator", "Jinji")
                    .addConfiguration("create_time",
                            System.currentTimeMillis() + "")
                    .build();

            admin.createNamespace(weibo);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 创建微博内容表
     * Table Name:weibo:content
     * RowKey:用户 ID_时间戳
     * ColumnFamily:info
     * ColumnLabel:标题 内容 图片 URL
     * Version:1 个版本
     */
    public void createTableContent() {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
//创建表表述
            HTableDescriptor content = new
                    HTableDescriptor(TableName.valueOf(TABLE_CONTENT));
//创建列族描述
            HColumnDescriptor info = new
                    HColumnDescriptor(Bytes.toBytes("info"));
//设置块缓存
            info.setBlockCacheEnabled(true);
//设置块缓存大小
            info.setBlocksize(2097152);
//设置压缩方式
// info.setCompressionType(Algorithm.SNAPPY);
//设置版本确界
            info.setMaxVersions(1);
            info.setMinVersions(1);
            content.addFamily(info);
            admin.createTable(content);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 用户关系表
     * Table Name:weibo:relations0
     * RowKey:用户 ID
     * ColumnFamily:attends,fans
     * ColumnLabel:关注用户 ID，粉丝用户 ID
     * ColumnValue:用户 ID
     * Version：1 个版本
     */
    public void createTableRelations() {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
            HTableDescriptor relations = new
                    HTableDescriptor(TableName.valueOf(TABLE_RELATIONS));
//关注的人的列族
            HColumnDescriptor attends = new
                    HColumnDescriptor(Bytes.toBytes("attends"));
//设置块缓存
            attends.setBlockCacheEnabled(true);
//设置块缓存大小
            attends.setBlocksize(2097152);
//设置压缩方式
// info.setCompressionType(Algorithm.SNAPPY);
//设置版本确界
            attends.setMaxVersions(1);
            attends.setMinVersions(1);
//粉丝列族
            HColumnDescriptor fans = new
                    HColumnDescriptor(Bytes.toBytes("fans"));
            fans.setBlockCacheEnabled(true);
            fans.setBlocksize(2097152);
            fans.setMaxVersions(1);
            fans.setMinVersions(1);
            relations.addFamily(attends);
            relations.addFamily(fans);
            admin.createTable(relations);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 创建微博收件箱表
     * Table Name: weibo:receive_content_email
     * RowKey:用户 ID
     * ColumnFamily:info
     * 尚硅谷大数据技术之 Hbase
     * —————————————————————————————
     * 更多 Java –大数据 –前端 –python 人工智能资料下载，可百度访问：尚硅谷官网
     * ColumnLabel:用户 ID-发布微博的人的用户 ID
     * ColumnValue:关注的人的微博的 RowKey
     * Version:1000
     */
    public void createTableReceiveContentEmail() {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
            HTableDescriptor receive_content_email = new
                    HTableDescriptor(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            HColumnDescriptor info = new
                    HColumnDescriptor(Bytes.toBytes("info"));
            info.setBlockCacheEnabled(true);
            info.setBlocksize(2097152);
            info.setMaxVersions(1000);
            info.setMinVersions(1000);
            receive_content_email.addFamily(info);
            ;
            admin.createTable(receive_content_email);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //增加微博增加内容
    public void publishContent(String uid, String content) {
        HConnection connection = null;
        try {
            connection = HConnectionManager.createConnection(conf);
//a、微博内容表中添加 1 条数据，首先获取微博内容表描述
            HTableInterface contentTBL =
                    connection.getTable(TableName.valueOf(TABLE_CONTENT));
//组装 Rowkey
            long timestamp = System.currentTimeMillis();
            String rowKey = uid + "_" + timestamp;
            Put put = new Put(Bytes.toBytes(rowKey));
            put.add(Bytes.toBytes("info"), Bytes.toBytes("content"),
                    timestamp, Bytes.toBytes(content));
            contentTBL.put(put);
//b、向微博收件箱表中加入发布的 Rowkey
//b.1、查询用户关系表，得到当前用户有哪些粉丝
            HTableInterface relationsTBL =
                    connection.getTable(TableName.valueOf(TABLE_RELATIONS));
//b.2、取出目标数据
            Get get = new Get(Bytes.toBytes(uid));
            get.addFamily(Bytes.toBytes("fans"));
            Result result = relationsTBL.get(get);
            List<byte[]> fans = new ArrayList<byte[]>();
//遍历取出当前发布微博的用户的所有粉丝数据
            for (Cell cell : result.rawCells()) {
                fans.add(CellUtil.cloneQualifier(cell));
            }
//如果该用户没有粉丝，则直接 return
            if (fans.size() <= 0) {
                return;
            }

//开始操作收件箱表
            HTableInterface recTBL =
                    connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL
                    ));
            List<Put> puts = new ArrayList<Put>();
            for (byte[] fan : fans) {
                Put fanPut = new Put(fan);
                fanPut.add(Bytes.toBytes("info"), Bytes.toBytes(uid),
                        timestamp, Bytes.toBytes(rowKey));
                puts.add(fanPut);
            }
            recTBL.put(puts);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != connection) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 关注用户逻辑
     * a、在微博用户关系表中，对当前主动操作的用户添加新的关注的好友
     * b、在微博用户关系表中，对被关注的用户添加粉丝（当前操作的用户）
     * c、当前操作用户的微博收件箱添加所关注的用户发布的微博 rowkey
     */
    public void addAttends(String uid, String... attends) {
//参数过滤
        if (attends == null || attends.length <= 0 || uid == null ||
                uid.length() <= 0) {
            return;
        }
        HConnection connection = null;
        try {
            connection = HConnectionManager.createConnection(conf);
//用户关系表操作对象（连接到用户关系表）
            HTableInterface relationsTBL =
                    connection.getTable(TableName.valueOf(TABLE_RELATIONS));
            List<Put> puts = new ArrayList<Put>();
//a、在微博用户关系表中，添加新关注的好友
            Put attendPut = new Put(Bytes.toBytes(uid));
            for (String attend : attends) {
//为当前用户添加关注的人
                attendPut.add(Bytes.toBytes("attends"),
                        Bytes.toBytes(attend), Bytes.toBytes(attend));
//b、为被关注的人，添加粉丝
                Put fansPut = new Put(Bytes.toBytes(attend));
                fansPut.add(Bytes.toBytes("fans"), Bytes.toBytes(uid),
                        Bytes.toBytes(uid));
//将所有关注的人一个一个的添加到 puts（List）集合中
                puts.add(fansPut);
            }
            puts.add(attendPut);
            relationsTBL.put(puts);
//c.1、微博收件箱添加关注的用户发布的微博内容（content）的 rowkey
            HTableInterface contentTBL =
                    connection.getTable(TableName.valueOf(TABLE_CONTENT));
            Scan scan = new Scan();
//用于存放取出来的关注的人所发布的微博的 rowkey
            List<byte[]> rowkeys = new ArrayList<byte[]>();
            for (String attend : attends) {
//过滤扫描 rowkey，即：前置位匹配被关注的人的 uid_
                RowFilter filter = new
                        RowFilter(CompareFilter.CompareOp.EQUAL, new
                        SubstringComparator(attend + "_"));
//为扫描对象指定过滤规则
                scan.setFilter(filter);
//通过扫描对象得到 scanner
                ResultScanner result = contentTBL.getScanner(scan);
//迭代器遍历扫描出来的结果集
                Iterator<Result> iterator = result.iterator();
                while (iterator.hasNext()) {
//取出每一个符合扫描结果的那一行数据
                    Result r = iterator.next();
                    for (Cell cell : r.rawCells()) {
//将得到的 rowkey 放置于集合容器中
                        rowkeys.add(CellUtil.cloneRow(cell));
                    }
                }
            }
//c.2、将取出的微博 rowkey 放置于当前操作用户的收件箱中
            if (rowkeys.size() <= 0) {
                return;
            }
//得到微博收件箱表的操作对象
            HTableInterface recTBL =
                    connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL
                    ));
//用于存放多个关注的用户的发布的多条微博 rowkey 信息
            List<Put> recPuts = new ArrayList<Put>();
            for (byte[] rk : rowkeys) {
                Put put = new Put(Bytes.toBytes(uid));
//uid_timestamp
                String rowKey = Bytes.toString(rk);
//借取 uid
                String attendUID = rowKey.substring(0,
                        rowKey.indexOf("_"));


                long timestamp =
                        Long.parseLong(rowKey.substring(rowKey.indexOf("_") + 1));
//将微博 rowkey 添加到指定单元格中
                put.add(Bytes.toBytes("info"), Bytes.toBytes(attendUID),
                        timestamp, rk);
                recPuts.add(put);
            }
            recTBL.put(recPuts);
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            if (null != connection) {
                try {
                    connection.close();
                } catch (IOException e) {
// TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 取消关注（remove)
     * a、在微博用户关系表中，对当前主动操作的用户删除对应取关的好友
     * b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人）
     * c、从收件箱中，删除取关的人的微博的 rowkey
     */
    public void removeAttends(String uid, String... attends) {
//过滤数据
        if (uid == null || uid.length() <= 0 || attends == null ||
                attends.length <= 0) {
            return;
        }
        HConnection connection = null;
        try {
            connection = HConnectionManager.createConnection(conf);
//a、在微博用户关系表中，删除已关注的好友
            HTableInterface relationsTBL =
                    connection.getTable(TableName.valueOf(TABLE_RELATIONS));
//待删除的用户关系表中的所有数据
            List<Delete> deletes = new ArrayList<Delete>();
//当前取关操作者的 uid 对应的 Delete 对象
            Delete attendDelete = new Delete(Bytes.toBytes(uid));
//遍历取关，同时每次取关都要将被取关的人的粉丝-1
            for (String attend : attends) {
                attendDelete.deleteColumn(Bytes.toBytes("attends"),

                        Bytes.toBytes(attend));
//b
                Delete fansDelete = new Delete(Bytes.toBytes(attend));
                fansDelete.deleteColumn(Bytes.toBytes("fans"),
                        Bytes.toBytes(uid));
                deletes.add(fansDelete);
            }
            deletes.add(attendDelete);
            relationsTBL.delete(deletes);
//c、删除取关的人的微博 rowkey 从 收件箱表中
            HTableInterface recTBL =
                    connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL
                    ));
            Delete recDelete = new Delete(Bytes.toBytes(uid));
            for (String attend : attends) {
                recDelete.deleteColumn(Bytes.toBytes("info"),
                        Bytes.toBytes(attend));
            }
            recTBL.delete(recDelete);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 获取微博实际内容
     * a、从微博收件箱中获取所有关注的人的发布的微博的 rowkey
     * b、根据得到的 rowkey 去微博内容表中得到数据
     * c、将得到的数据封装到 Message 对象中
     */
    public List<Message> getAttendsContent(String uid) {
        HConnection connection = null;
        try {
            connection = HConnectionManager.createConnection(conf);
            HTableInterface recTBL =
                    connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL
                    ));
//a、从收件箱中取得微博 rowKey
            Get get = new Get(Bytes.toBytes(uid));
//设置最大版本号
            get.setMaxVersions(5);
            List<byte[]> rowkeys = new ArrayList<byte[]>();
            Result result = recTBL.get(get);
            for (Cell cell : result.rawCells()) {
                rowkeys.add(CellUtil.cloneValue(cell));
            }
//b、根据取出的所有 rowkey 去微博内容表中检索数据
            HTableInterface contentTBL = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            List<Get> gets = new ArrayList<>();
//根据 rowkey 取出对应微博的具体内容
            for (byte[] rk : rowkeys) {
                Get g = new Get(rk);
                gets.add(g);
            }
//得到所有的微博内容的 result 对象
            Result[] results = contentTBL.get(gets);
            List<Message> messages = new ArrayList<Message>();
            for (Result res : results) {
                for (Cell cell : res.rawCells()) {
                    Message message = new Message();
                    String rowKey =
                            Bytes.toString(CellUtil.cloneRow(cell));
                    String userid = rowKey.substring(0,
                            rowKey.indexOf("_"));
                    String timestamp =
                            rowKey.substring(rowKey.indexOf("_") + 1);
                    String content =
                            Bytes.toString(CellUtil.cloneValue(cell));
                    message.setContent(content);
                    message.setTimestamp(timestamp);
                    message.setUid(userid);
                    messages.add(message);
                }
            }
            return messages;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }
}